package com.niodata.dp.web.module.task;

import com.fasterxml.jackson.annotation.JsonIgnore;
import com.google.common.base.Strings;
import com.niodata.dp.core.exception.InvalidParamException;
import com.niodata.dp.core.usermgr.model.UserInfo;
import com.niodata.dp.log.LogUtil;
import com.niodata.dp.log.viewer.ViewerSession;
import com.niodata.dp.task.dao.JobLogMappingMapper;
import com.niodata.dp.task.entity.JobLogMapping;
import com.niodata.dp.task.log.LogType;
import com.niodata.dp.task.log.LogViewMode;
import com.niodata.dp.task.log.TaskLogManager;
import com.niodata.dp.web.module.base.BaseController;
import com.niodata.dp.web.security.ContextUtil;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import javax.servlet.RequestDispatcher;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController("LogConsoleController")
@RequestMapping("/v1/task/log")
public class LogConsoleController extends BaseController {

  private static final Logger logger = org.apache.log4j.LogManager.getLogger(
        LogConsoleController.class);
  private final int logDataWaitTime = 1000;
  @Autowired
  JobLogMappingMapper jobLogMapper;
  @Autowired
  @Qualifier("taskLogManager")
  TaskLogManager logManager;

  @Autowired
  LogConsoleManager logConsoleManager;

  /**
   * spark ui.
   */
  @RequestMapping("sparkui")
  public void sparkUi(@RequestParam("execId") String execId,
        @RequestParam("jobName") String jobName,
        @RequestParam("logType") LogType logType, HttpServletRequest
        request, HttpServletResponse response)
        throws ServletException, IOException {

    JobLogMapping jobLogInfo = jobLogMapper.getJobLogMapping(execId, jobName, logType);
    if (jobLogInfo != null && !Strings.isNullOrEmpty(jobLogInfo.getYarnAppId())) {
      RequestDispatcher dispatcher = request.getRequestDispatcher("/proxy/"
            + jobLogInfo.getYarnAppId());
      dispatcher.forward(request, response);
    } else {
      logger.error("yarn app not found ");
    }
  }

  @RequestMapping("openConsole")
  public Object openLogConsole(@RequestParam("taskId") String taskId,
        @RequestParam("execId") int execId,
        @RequestParam("jobName") String jobName) {
    LogConsole console = new LogConsole();
    String consoleId = UUID.randomUUID().toString();
    console.setId(consoleId);
    console.setCreateTime(new Date());
    console.setExecId(execId);
    console.setTaskId(taskId);
    console.setJobName(jobName);

    logManager.checkJobInfo(taskId, jobName);

    UserInfo userInfo = ContextUtil.getCurrentUser();
    console.setUser(userInfo.getUserName());
    logConsoleManager.put(consoleId, console);
    logger.info("open log console [" + consoleId + "]" + execId);
    return getResult(console);
  }

  @RequestMapping("closeConsole")
  public Object closeLogConsole(@RequestParam("id") String consoleId) {
    LogConsole console = logConsoleManager.get(consoleId);
    try {
      if (console != null) {
        console.viewerSession.close();
        console.viewerSession.clear();
      }
    } catch (Exception e) {
      LogUtil.logStackTrace(logger, e);
    } finally {
      logConsoleManager.remove(consoleId);
    }
    return getResult(true);
  }

  /**
   * connect log file.
   */
  @RequestMapping("connect")
  public Object connectLogFile(@RequestParam("id") String consoleId,
        @RequestParam("viewMode") LogViewMode viewMode,
        @RequestParam("logType") LogType logType,
        @RequestParam("connectParam") String connectParam) throws IOException {
    LogConsole console = logConsoleManager.get(consoleId);
    if (null == console) {
      throw new InvalidParamException("logconsole not exists");
    }
    ViewerSession session = logManager.connectLog(console.getTaskId(), console.getExecId(),
          console.getJobName(),
          viewMode,
          connectParam,
          logType);
    if (console.viewerSession != null) {
      console.viewerSession.close();
      console.viewerSession.clear();
    }
    console.viewerSession = session;
    session.open();
    logConsoleManager.put(consoleId, console);
    return getResult(console);
  }

  @RequestMapping("nextPage")
  public Object nextPageLog(@RequestParam("id") String consoleId) {
    LogConsole console = logConsoleManager.get(consoleId);
    Map<String, Object> result = new HashMap<>();
    List lines = console.viewerSession.readLines(100, logDataWaitTime);
    result.put("lines", lines);
    result.put("hasMore", console.viewerSession.hasMore());
    return getResult(result);
  }

  class LogConsole {

    @JsonIgnore
    volatile ViewerSession viewerSession;
    private String id;
    private String user;
    private Date createTime;
    private String taskId;
    private String jobName;
    private int execId;
    private LogType logType;
    private LogViewMode viewMode;

    public void connect() {

    }

    public String getId() {
      return id;
    }

    public void setId(String id) {
      this.id = id;
    }

    public String getUser() {
      return user;
    }

    public void setUser(String user) {
      this.user = user;
    }

    public Date getCreateTime() {
      return createTime;
    }

    public void setCreateTime(Date createTime) {
      this.createTime = createTime;
    }

    public String getTaskId() {
      return taskId;
    }

    public void setTaskId(String taskId) {
      this.taskId = taskId;
    }

    public String getJobName() {
      return jobName;
    }

    public void setJobName(String jobName) {
      this.jobName = jobName;
    }

    public int getExecId() {
      return execId;
    }

    public void setExecId(int execId) {
      this.execId = execId;
    }
  }
}